package com.xquant.common.engine.xPP.cxf;


import com.xquant.common.engine.xPP.cxf.webservice.NotifyParam;
import org.apache.commons.collections.MapUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Service("XQuantPPServiceCxf")
@EnableScheduling
public class XQuantPPService {

    private static final Lock lock = new ReentrantLock();
    private ExecutorService notifyExec = null;
    private static final String XQUANT_LOCK_TYPE_SHICK = "shick";
    private static final String XQUANT_LOCK_TYPE_FINE = "fine";
    private Object obj = new Object();
    private List<Map> pp = new ArrayList<>();

    public List<Map> getPp() {
        return pp;
    }

    public void setPp(List<Map> pp) {
        this.pp = pp;
    }

    private List<XQuantPPStubFactory> lst = new ArrayList<>();

    public List<XQuantPPStubFactory> getLst() {
        return lst;
    }

    private int index = 0;

    private int addressIndex = 0;

    private Long timeOut = 10000L;

    private int maxCycleNum = 3;

    private String localPath;

    public String getLocalPath() {
        return localPath;
    }

    private static BlockingQueue<XQuantPPStubFactory> stubFactoryQueue = new LinkedBlockingDeque();

    public void setLocalPath(String localPath) {
        this.localPath = localPath;
    }

    private List<String> uploadDllNames = new ArrayList<>();

    public List<String> getUploadDllNames() {
        return uploadDllNames;
    }

    @Autowired
    public XQuantPPConfig xQuantPPConfig;

    @PostConstruct
    private void init() {

        Map ppMap = new HashMap();
        ppMap.put("connectionTimeout", xQuantPPConfig.getConnectionTimeout());
        ppMap.put("receiveTimeout", xQuantPPConfig.getReceiveTimeout());
        ppMap.put("threads", xQuantPPConfig.getThreads());
        String[] hostPorts = xQuantPPConfig.getHostPort();
        for (int i = 0; i < hostPorts.length; i++) {
            StringBuilder sb = new StringBuilder("http://" + hostPorts[i] + "/axis2/services/xQuantPP");
            ppMap.put("address", sb.toString());
            ppMap.put("status", "-1");
            lst.add(new XQuantPPStubFactory(ppMap));
        }
        testAll();
    }

    @Scheduled(fixedDelay = 60000)
    public void testAll() {

        ExecutorService exec = Executors.newFixedThreadPool(lst.size());
        for (XQuantPPStubFactory item : lst) {
            exec.execute(new Runnable() {
                @Override
                public void run() {
                    Integer status = item.getStatus();
                   /* if (item.testMethod(item) && status < 0) {
                        stubFactoryQueue.add(item);
                    }*/
                }
            });
        }
        try {
            exec.shutdown();
            exec.awaitTermination(1, TimeUnit.HOURS);
        } catch (Exception ex) {
            //ex.printStackTrace();
        }
    }

    /**
     * @param address
     * @return -1异常，0正常，1无此计算服务
     */
    public int test(String address) {
        for (XQuantPPStubFactory item : lst) {
            if (item.getxQuantPPStub().getAddress().equals(address)) {
                if (item.testMethod(item)) {
                    if (item.getStatus() < 0) {
                        stubFactoryQueue.add(item);
                    }
                    return 0;
                }
                return -1;
            }
        }
        return 1;
    }

    public void restartAll() {
        ExecutorService exec = Executors.newFixedThreadPool(lst.size());
        for (XQuantPPStubFactory item : lst) {
            if (MapUtils.getInteger(item.getPp(), "status") >= 0) {
                exec.execute(new Runnable() {
                    @Override
                    public void run() {
                        item.restart();
                    }
                });
            }
        }
        try {
            exec.shutdown();
            exec.awaitTermination(1, TimeUnit.HOURS);
        } catch (Exception ex) {

        }
    }

    /**
     * @param address
     * @return -1异常，0正常，1无此计算服务
     */
    public int restart(String address) {
        for (XQuantPPStubFactory item : lst) {
            if (item.getxQuantPPStub().getAddress().equals(address)) {
                try {
                    item.restart();
                    return 0;
                } catch (Exception e) {
                    e.printStackTrace();
                    return -1;
                }
            }
        }
        return 1;
    }

    public XQuantPPStub borrowStub() throws Exception {
        /*if (XQUANT_LOCK_TYPE_FINE.equals(ParamCache.getString("xquant.lock.type"))) {
            return borrowStub4Fine();
        } else {*/
            return borrowStub4Shick();
//        }
    }

    public XQuantPPStub borrowStubServer() throws Exception {
        lock.lock();
        try {
            while (true) {
                for (int i = 0; i < lst.size(); i++) {
                    index = (1 + index);
                    if (index >= lst.size() * 5) {
                        index = 0;
                    }
                    int id = index / 5;
                    if (lst.get(id).getStatus() >= 0) {
                        XQuantPPStub stub = lst.get(id).getxQuantPPStub();
                        if (stub.getQueue().get() < stub.getThreads()) {
                            stub.getQueue().incrementAndGet();
                            return stub;
                        }
                    }
                }
            }
        } finally {
            lock.unlock();
        }
    }

    public XQuantPPStub borrowStub4Shick() throws Exception {
        lock.lock();
        try {
            Long start = System.currentTimeMillis();
            while (true) {
                for (int i = 0; i < lst.size(); i++) {
                    index = (index + 1) % lst.size();
                    if (System.currentTimeMillis() - start >= timeOut) {
                        //所有计算服务都不可用时，status永远<0，线程阻塞导致死锁
                        throw new RuntimeException("计算服务不可用");
                    }
                    if (lst.get(index).getStatus() >= 0) {
                        XQuantPPStub stub = lst.get(index).getxQuantPPStub();
                        if (stub.getQueue().get() < stub.getThreads()) {
                            stub.getQueue().incrementAndGet();
                            return stub;
                        }
                    }
                }
            }
        } finally {
            lock.unlock();
        }
    }


    public XQuantPPStub borrowStub4Fine() throws Exception {
        int cycleNum = 0;
        while (cycleNum < maxCycleNum) {
            XQuantPPStubFactory stubFactory = null;
            lock.lock();
            try {
                while (true) {
                    stubFactory = stubFactoryQueue.poll();
                    if (stubFactory != null) {
                        break;
                    }
                }
            } finally {
                lock.unlock();
            }
            stubFactory.getLock().lock();
            try {
                Long start = System.currentTimeMillis();
                while (true) {
                    if (stubFactory.getStatus() >= 0) {
                        XQuantPPStub stub = stubFactory.getxQuantPPStub();
                        if (stub.getQueue().get() < stub.getThreads()) {
                            stub.getQueue().incrementAndGet();
                            stubFactoryQueue.add(stubFactory);
                            return stub;
                        }
                    } else {
                        break;
                    }
                }
            } finally {
                stubFactory.getLock().unlock();
            }
        }
        throw new RuntimeException("无可用计算服务");
    }

    public XQuantPPStub borrowStubPriority() throws Exception {
        for (int i = 0; i < lst.size(); i++) {
            index = (index + 1) % lst.size();
            if (MapUtils.getInteger(lst.get(index).getPp(), "status") >= 0) {
                XQuantPPStub stub = lst.get(index).getxQuantPPStub();
                stub.getQueue().incrementAndGet();
                return stub;
            }
        }
        throw new RuntimeException("没有可用的计算服务");
    }

    public void returnStub(XQuantPPStub stub) {
        stub.getQueue().decrementAndGet();
    }

    public void setTimeOut(XQuantPPStub stub) {
        String address = stub.getAddress();
        for (int i = 0; i < lst.size(); i++) {
            if (address.equals(MapUtils.getString(lst.get(i).getPp(), "address"))) {
                lst.get(i).setStatus(-1);
                return;
            }
        }
    }

    public Map getPPItem() {
        for (int i = 0; i < lst.size(); i++) {
            addressIndex = (addressIndex + 1) % lst.size();
            if (MapUtils.getInteger(lst.get(addressIndex).getPp(), "status") >= 0) {
                return lst.get(addressIndex).getPp();
            }
        }
        return null;
    }

    public void notifyUploadDll() {
        ExecutorService exec = Executors.newFixedThreadPool(lst.size());
        for (XQuantPPStubFactory item : lst) {
            if (MapUtils.getObject(item.getPp(), "uploadDllNames") == null) {
                item.getPp().put("uploadDllNames", new ArrayList<String>());
            }
            ((List) item.getPp().get("uploadDllNames")).addAll(uploadDllNames);
            if (MapUtils.getInteger(item.getPp(), "status") >= 0) {
                exec.execute(new Runnable() {
                    @Override
                    public void run() {
                        item.uploadDll();
                    }
                });
            }
        }
        try {
            exec.shutdown();
//            exec.awaitTermination(1, TimeUnit.HOURS);
            uploadDllNames.clear();
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

    public int getThreads() {
        int ret = 0;
        for (int i = 0; i < lst.size(); i++) {
            if (lst.get(i).getStatus() >= 0) {
                ret += lst.get(i).getxQuantPPStub().getThreads();
            }
        }
        return ret;
    }

    public void notifyData(NotifyParam param) {
        if (notifyExec == null) {
            notifyExec = Executors.newFixedThreadPool(lst.size());
        }
        for (XQuantPPStubFactory item : lst) {
            try {
                if (MapUtils.getInteger(item.getPp(), "status") >= 0) {
                    notifyExec.execute(new Runnable() {
                        @Override
                        public void run() {
                            item.notifyData(param);
                        }
                    });
                }
            } catch (Exception e) {
                //e.printStackTrace();
            }
        }
    }
}
